package com.idanchuang.component.mq.amqp;

import com.idanchuang.component.mq.amqp.config.ComponentAmqpConfig;
import com.idanchuang.component.mq.amqp.processor.AfterReceivePostProcessor;
import com.idanchuang.component.mq.amqp.processor.BeforePublishPostProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;

/**
 * @author yjy
 * @date 2019/11/8 11:12
 **/
@Configuration
public class AmqpAutoConfiguration {

    private static final Logger rabbitTemplateLogger = LoggerFactory.getLogger(RabbitTemplate.class);
    private static final Logger abstractRabbitListenerContainerFactoryLogger = LoggerFactory.getLogger(AbstractRabbitListenerContainerFactory.class);

    private final Field beforePublishPostProcessorsField;
    private final Field afterReceivePostProcessorsField;

    public AmqpAutoConfiguration() {
        Field beforePublishPostProcessorsField = null;
        try {
            beforePublishPostProcessorsField = RabbitTemplate.class.getDeclaredField("beforePublishPostProcessors");
            beforePublishPostProcessorsField.setAccessible(true);
        } catch (NoSuchFieldException e) {
            // ignore
        }
        this.beforePublishPostProcessorsField = beforePublishPostProcessorsField;
        Field afterReceivePostProcessorsField = null;
        try {
            afterReceivePostProcessorsField = AbstractRabbitListenerContainerFactory.class.getDeclaredField("afterReceivePostProcessors");
            afterReceivePostProcessorsField.setAccessible(true);
        } catch (NoSuchFieldException e) {
            // ignore
        }
        this.afterReceivePostProcessorsField = afterReceivePostProcessorsField;
    }

    @Bean
    @ConfigurationProperties("spring.rabbitmq.amqp")
    public ComponentAmqpConfig componentAmqpConfig() {
        return new ComponentAmqpConfig();
    }

    @Bean
    @ConditionalOnProperty(prefix = "spring.rabbitmq.amqp", name = "aliyun", havingValue = "true", matchIfMissing = false)
    @ConditionalOnMissingBean(AliyunCredentialsProvider.class)
    public AliyunCredentialsProvider credentialsProvider(RabbitProperties rabbitProperties, ComponentAmqpConfig amqpConfig) {
        return new AliyunCredentialsProvider(rabbitProperties, amqpConfig);
    }

    @Bean
    @ConditionalOnBean(AliyunCredentialsProvider.class)
    @ConditionalOnMissingBean(ConnectionFactory.class)
    public ConnectionFactory getConnectionFactory(
            RabbitProperties rabbitProperties, AliyunCredentialsProvider credentialsProvider) {
        com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory =
                new com.rabbitmq.client.ConnectionFactory();
        rabbitConnectionFactory.setHost(rabbitProperties.getHost());
        rabbitConnectionFactory.setPort(rabbitProperties.getPort());
        rabbitConnectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
        rabbitConnectionFactory.setCredentialsProvider(credentialsProvider);
        rabbitConnectionFactory.setAutomaticRecoveryEnabled(true);
        rabbitConnectionFactory.setNetworkRecoveryInterval(5000);
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
        connectionFactory.setPublisherConfirms(rabbitProperties.isPublisherConfirms());
        connectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns());
        return connectionFactory;
    }

    @Bean
    @ConditionalOnMissingBean(name = "configBeforePublishPostProcessors")
    public Object configBeforePublishPostProcessors(List<RabbitTemplate> rabbitTemplates, List<BeforePublishPostProcessor> beforePublishPostProcessors) {
        if (rabbitTemplates == null || rabbitTemplates.isEmpty() || beforePublishPostProcessors.isEmpty()) {
            return new Object();
        }
        for (RabbitTemplate rabbitTemplate : rabbitTemplates) {
            before(rabbitTemplate, beforePublishPostProcessors);
        }
        return new Object();
    }


    @Bean
    @ConditionalOnMissingBean(name = "configAfterReceivePostProcessors")
    public Object configAfterReceivePostProcessors(List<AbstractRabbitListenerContainerFactory> containerFactories, List<AfterReceivePostProcessor> afterReceivePostProcessors) {
        if (containerFactories == null || containerFactories.isEmpty() || afterReceivePostProcessors.isEmpty()) {
            return new Object();
        }
        for (AbstractRabbitListenerContainerFactory factory : containerFactories) {
            after(factory, afterReceivePostProcessors);
        }
        return new Object();
    }

    /**
     * 发送拦截
     * @param rabbitTemplate ..
     */
    private void before(RabbitTemplate rabbitTemplate, List<BeforePublishPostProcessor> beforePublishPostProcessors) {
        rabbitTemplateLogger.info("configPressureAmqpTemplate > rabbitTemplate: {}", rabbitTemplate);
        if (beforePublishPostProcessorsField == null) {
            return;
        }
        Collection<MessagePostProcessor> processors;
        try {
            processors = (Collection) beforePublishPostProcessorsField.get(rabbitTemplate);
        } catch (IllegalAccessException e) {
            return;
        }

        // If there are no existing post processors, return only the one
        if (processors == null) {
            rabbitTemplate.setBeforePublishPostProcessors(beforePublishPostProcessors.toArray(new MessagePostProcessor[0]));
            return;
        }

        List<MessagePostProcessor> befores = new ArrayList<>(processors);
        for (BeforePublishPostProcessor beforePublishPostProcessor : beforePublishPostProcessors) {
            for (MessagePostProcessor processor : befores) {
                if (beforePublishPostProcessor.getClass().isAssignableFrom(processor.getClass())) {
                    break;
                }
            }
            befores.add(beforePublishPostProcessor);
        }
        rabbitTemplate.setBeforePublishPostProcessors(befores.toArray(new MessagePostProcessor[0]));
    }

    /**
     * 消费拦截
     * @param containerFactory ..
     */
    private void after(AbstractRabbitListenerContainerFactory containerFactory, List<AfterReceivePostProcessor> afterReceivePostProcessors) {
        abstractRabbitListenerContainerFactoryLogger.info("configPressureAmqpTemplate > factory: {}", containerFactory);
        if (afterReceivePostProcessorsField == null) {
            return;
        }
        MessagePostProcessor[] processors;
        try {
            processors = (MessagePostProcessor[]) afterReceivePostProcessorsField.get(containerFactory);
        } catch (IllegalAccessException e) {
            return;
        }

        // If there are no existing post processors, return only the one
        if (processors == null) {
            containerFactory.setAfterReceivePostProcessors(afterReceivePostProcessors.toArray(new MessagePostProcessor[0]));
            return;
        }

        List<MessagePostProcessor> afters = new ArrayList<>(Arrays.asList(processors));
        for (AfterReceivePostProcessor afterReceivePostProcessor : afterReceivePostProcessors) {
            for (MessagePostProcessor processor : afters) {
                if (afterReceivePostProcessor.getClass().isAssignableFrom(processor.getClass())) {
                    break;
                }
            }
            afters.add(afterReceivePostProcessor);
        }

        containerFactory.setAfterReceivePostProcessors(afters.toArray(new MessagePostProcessor[0]));
    }

}
